package com.hdu.dwh.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hdu.dwh.dtos.PageRequestDto;
import com.hdu.dwh.dtos.SourceAccessDto;
import com.hdu.dwh.mapper.SourceAccessMapper;
import com.hdu.dwh.pojos.source_access_pojos.SourceAccess;
import com.hdu.dwh.pojos.source_access_pojos.IColumn;
import com.hdu.dwh.pojos.source_access_pojos.SourceAccessDatabase;
import com.hdu.dwh.pojos.source_access_pojos.SourceAccessDetails;
import com.hdu.dwh.pojos.source_access_pojos.SourceAccessTable;
import com.hdu.dwh.service.ISourceAccessService;
import com.hdu.dwh.utils.ExcludeDatabaseUtils;
import com.hdu.dwh.utils.OperatorUtil;
import com.hdu.dwh.vos.SourceAccessVo;

import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoClients;
import com.mongodb.client.MongoDatabase;
import com.mongodb.client.MongoIterable;
import lombok.extern.slf4j.Slf4j;

import com.hdu.dwh.constants.CommonConstants;
import com.hdu.dwh.results.PageResponseResult;
import com.hdu.dwh.results.ResponseResult;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import java.sql.*;
import java.util.HashSet;
import java.util.List;
import java.util.Properties;
import java.util.Set;

/**
 * <p>
 * 数据接入表 服务实现类
 * </p>
 *
 * @author bobo
 * @since 2025-04-08
 */
@Service
@Slf4j
public class SourceAccessServiceImpl extends ServiceImpl<SourceAccessMapper, SourceAccess>
        implements ISourceAccessService {

    /**
     * 添加接入信息
     * 
     * @param sourceAccessDto
     * @return
     */
    @Override
    public ResponseResult add(SourceAccessDto sourceAccessDto) {
        if (sourceAccessDto == null) {
            return ResponseResult.fail(CommonConstants.PARAM_IS_NULL);
        }
        SourceAccess sourceAccess = new SourceAccess().setCreatedBy(OperatorUtil.getCurrentOperator()); //添加创建用户信息
        BeanUtils.copyProperties(sourceAccessDto, sourceAccess);
        // todo 增加用户信息
        // 自己做连接名的唯一处理
        LambdaQueryWrapper<SourceAccess> queryWrapper = new LambdaQueryWrapper<SourceAccess>()
                .eq(SourceAccess::getConnectionName, sourceAccessDto.getConnectionName())
                .eq(SourceAccess::getIsDeleted, false);
        if (count(queryWrapper) != 0) {
            return ResponseResult.fail(CommonConstants.CONNECTION_NAME_EXIST);
        }
        if (!save(sourceAccess)) {
            return ResponseResult.fail(CommonConstants.ADD_FAILED);
        }
        return ResponseResult.success(CommonConstants.ADD_SUCCESS);
    }

    /**
     * 修改接入信息
     * 
     * @param sourceAccessDto
     * @param id
     * @return
     */
    @Override
    public ResponseResult edit(SourceAccessDto sourceAccessDto, Long id) {
        if (sourceAccessDto == null) {
            return ResponseResult.fail(CommonConstants.PARAM_IS_NULL);
        }
        if (id == null) {
            return ResponseResult.fail(CommonConstants.PARAM_IS_NULL);
        }
        SourceAccess sourceAccess = new SourceAccess();
        BeanUtils.copyProperties(sourceAccessDto, sourceAccess);
        sourceAccess.setId(id);
        if (!updateById(sourceAccess)) {
            return ResponseResult.fail(CommonConstants.EDIT_FAILED);
        }
        return ResponseResult.success(CommonConstants.EDIT_SUCCESS);
    }

    /**
     * 删除接入信息
     * 
     * @param id
     * @return
     */
    @Override
    public ResponseResult delete(Long id) {
        if (id == null) {
            return ResponseResult.fail(CommonConstants.PARAM_IS_NULL);
        }
        if (!removeById(id)) {
            return ResponseResult.fail(CommonConstants.DELETE_FAILED);
        }
        return ResponseResult.success(CommonConstants.DELETE_SUCCESS);
    }

    @Override
    public ResponseResult pageQuery(PageRequestDto pageRequestDto) {
        if (pageRequestDto == null) {
            return PageResponseResult.fail(CommonConstants.PARAM_IS_NULL);
        }
        pageRequestDto.checkParam();
        // 分页查询
        IPage<SourceAccess> page = new Page<>(pageRequestDto.getPage(), pageRequestDto.getSize());
        LambdaQueryWrapper<SourceAccess> wrapper = new LambdaQueryWrapper<SourceAccess>()
                .orderByAsc(SourceAccess::getCreatedAt);
        page = page(page, wrapper);
        // 封装vo
        List<SourceAccessVo> voList = page.getRecords().stream()
                .map(entity -> {
                    SourceAccessVo vo = new SourceAccessVo();
                    BeanUtils.copyProperties(entity, vo);
                    return vo;
                }).toList();
        // 返回
        if (page.getTotal() == 0) {
            return PageResponseResult.fail(CommonConstants.RECORD_NOT_EXIST);
        }
        return PageResponseResult.success((int) page.getCurrent(), (int) page.getPages(), (int) page.getSize(),
                (int) page.getTotal(), voList);
    }

    @Override
    public SourceAccessDetails getAll_dbs_tables_columnsFromSourceAccess(SourceAccess sourceAccess) throws Exception {
        String sourceType = sourceAccess.getSourceType();
        // 如果是MongoDB，需要特殊处理
        if ("mongodb".equalsIgnoreCase(sourceType)) {
            SourceAccessDetails sourceAccessDetails = new SourceAccessDetails();
            sourceAccessDetails.setSourceAccess(sourceAccess);
            sourceAccessDetails.setDatabases(this.getMongoDBDatabaseDetails(sourceAccess));
            return sourceAccessDetails;
        }

        Connection connection = this.createConnectionFromSourceAccess(sourceAccess);

        SourceAccessDetails sourceAccessDetails = new SourceAccessDetails();
        sourceAccessDetails.setSourceAccess(sourceAccess);
        // 对PostgreSQL进行特殊处理，获取正确的数据库列表
        if ("postgresql".equalsIgnoreCase(sourceType)) {
            sourceAccessDetails.setDatabases(this.getPostgreSQLDatabaseDetails(connection, sourceAccess));
        } else {
            sourceAccessDetails.setDatabases(this.getAllDatabaseDetails(connection, null));
        }

        connection.close();
        return sourceAccessDetails;
    }

    /**
     * 根据SourceAccess信息创建数据库连接
     *
     * @param SourceAccess {@link SourceAccess}
     * @return Connection {@link Connection}
     * @throws Exception
     */
    private Connection createConnectionFromSourceAccess(SourceAccess sourceAccess) throws Exception {

        String sourceType = sourceAccess.getSourceType();

        // 如果是MongoDB，需要特殊处理
        if ("mongodb".equalsIgnoreCase(sourceType)) {
            // MongoDB使用MongoClient而不是JDBC连接
            // 这里需要MongoDB特定的连接逻辑
            testMongoDBConnection(sourceAccess);
            // MongoDB连接测试成功后，返回null或者抛出特殊异常
            throw new UnsupportedOperationException("MongoDB为NoSQL数据库，不支持传统JDBC元数据获取");
        }
        String host = sourceAccess.getHost();
        Integer port = sourceAccess.getPort();
//        String sourceType = sourceAccess.getSourceType();
        String defaultDatabase = sourceAccess.getDefaultDatabase();
        String jdbcUrl = String.format("jdbc:%s://%s:%d/%s", sourceType, host, port, defaultDatabase);
        System.out.println(jdbcUrl);
        String username = sourceAccess.getUsername();
        String password = sourceAccess.getPassword();
        Connection connection = createConnection(jdbcUrl, username, password);
        return connection;
    }

    /**
     * 创建数据库连接
     * 
     * @param jdbcUrl
     * @param username
     * @param password
     * @return
     * @throws Exception
     */
    private Connection createConnection(String jdbcUrl, String username, String password) throws Exception {
        Properties properties = new Properties();
        properties.setProperty("user", username);
        if (password != null) {
            properties.setProperty("password", password);
        }
        properties.setProperty("SSL", "false");

        Connection connection = DriverManager.getConnection(jdbcUrl, properties);
        return connection;
    }

    /**
     * 返回所有数据库的详细信息，包括表和列
     *
     * @param connection 数据库连接（关系型数据库使用）
     * @param sourceType 数据源类型
     * @return Set<SourceAccessDatabase> {@link SourceAccessDatabase}
     * @throws SQLException SQL异常
     */
    @Override
    public Set<SourceAccessDatabase> getAllDatabaseDetails(Connection connection, String catalogName)
            throws SQLException {
        // TODO: 可能需要重构，方法过于复杂
        Set<SourceAccessDatabase> databases = new HashSet<>();
        DatabaseMetaData metaData = connection.getMetaData();

        // 首先获取所有数据库/模式
        ResultSet schemas = metaData.getSchemas();
        while (schemas.next()) {
            String databaseName = schemas.getString("TABLE_SCHEM");
            if (databaseName != null && !ExcludeDatabaseUtils.isExcludeDatabase(databaseName)) {
                SourceAccessDatabase database = new SourceAccessDatabase();
                database.setDatabaseName(databaseName);
                database.setTables(new HashSet<>());
                databases.add(database);
            }
        }

        // 如果没有找到schemas，尝试使用catalogs
        if (databases.isEmpty()) {
            ResultSet catalogs = metaData.getCatalogs();
            while (catalogs.next()) {
                String t_catalogName = catalogs.getString("TABLE_CAT");

                if (t_catalogName != null && !ExcludeDatabaseUtils.isExcludeDatabase(t_catalogName)) {
                    log.info("发现数据库: " + t_catalogName);
                    SourceAccessDatabase database = new SourceAccessDatabase();
                    database.setDatabaseName(t_catalogName);
                    database.setTables(new HashSet<>());
                    databases.add(database);
                }
            }
        }

        // 然后获取所有列信息，并关联到对应的数据库和表
        ResultSet columns = metaData.getColumns(catalogName, "%", "%", null);
        while (columns.next()) {
            String tableCat = columns.getString("TABLE_CAT");
            String tableSchem = columns.getString("TABLE_SCHEM");
            String tableName = columns.getString("TABLE_NAME");
            String columnName = columns.getString("COLUMN_NAME");
            String columnType = columns.getString("TYPE_NAME");

            // 排除系统数据库
            if (tableCat != null && tableCat.contains("schema")) {
                continue;
            }

            String databaseName = tableSchem != null ? tableSchem : tableCat;

            // 查找对应的数据库对象
            SourceAccessDatabase database = databases.stream()
                    .filter(db -> db.getDatabaseName().equals(databaseName))
                    .findFirst()
                    .orElse(null);

            // 如果没找到数据库，创建一个新的
            if (database == null ) {
                continue;
            }

            // 查找或创建表对象
            SourceAccessTable table = database.getTables().stream()
                    .filter(t -> t.getTableName().equals(tableName))
                    .findFirst()
                    .orElse(null);

            if (table == null) {
                table = new SourceAccessTable();
                table.setTableName(tableName);
                table.setColumns(new HashSet<>());
                database.getTables().add(table);
            }

            // 创建列对象
            IColumn column = new IColumn();
            column.setColumnName(columnName);
            column.setColumnType(columnType);
            table.getColumns().add(column);
        }

        log.info("查询结束");
        return databases;
    }

    /**
     * 测试MongoDB连接
     *
     * @param sourceAccess MongoDB连接信息
     * @throws Exception 连接异常
     */
    private void testMongoDBConnection(SourceAccess sourceAccess) throws Exception {
        String host = sourceAccess.getHost();
        Integer port = sourceAccess.getPort();
        String username = sourceAccess.getUsername();
        String password = sourceAccess.getPassword();
        String databaseName = sourceAccess.getDefaultDatabase();

        String connectionString;
        if (username != null && !username.isEmpty() && password != null && !password.isEmpty()) {
            connectionString = String.format("mongodb://%s:%s@%s:%d/%s", username, password, host, port, databaseName);
        } else {
            connectionString = String.format("mongodb://%s:%d/%s", host, port, databaseName);
        }

        try (MongoClient mongoClient = MongoClients.create(connectionString)) {
            // 测试连接是否有效
            MongoDatabase database = mongoClient.getDatabase(databaseName);
            // 尝试获取集合列表来验证连接
            MongoIterable<String> collections = database.listCollectionNames();
            // 至少尝试访问一个元素来验证连接
            collections.first();
        } catch (Exception e) {
            throw new Exception("MongoDB连接测试失败: " + e.getMessage(), e);
        }
    }
    /**
     * 获取MongoDB数据库详细信息
     *
     * @param sourceAccess MongoDB连接信息
     * @return Set<SourceAccessDatabase> MongoDB数据库信息集合
     * @throws Exception 连接或查询异常
     */
    private Set<SourceAccessDatabase> getMongoDBDatabaseDetails(SourceAccess sourceAccess) throws Exception {
        Set<SourceAccessDatabase> databases = new HashSet<>();

        String host = sourceAccess.getHost();
        Integer port = sourceAccess.getPort();
        String username = sourceAccess.getUsername();
        String password = sourceAccess.getPassword();
        String databaseName = sourceAccess.getDefaultDatabase();

        String connectionString;
        if (username != null && !username.isEmpty() && password != null && !password.isEmpty()) {
            connectionString = String.format("mongodb://%s:%s@%s:%d/%s", username, password, host, port, databaseName);
        } else {
            connectionString = String.format("mongodb://%s:%d/%s", host, port, databaseName);
        }

        try (MongoClient mongoClient = MongoClients.create(connectionString)) {
            // 获取所有数据库名称
            MongoIterable<String> databaseNames = mongoClient.listDatabaseNames();
            for (String dbName : databaseNames) {
                // 跳过系统数据库
                if (ExcludeDatabaseUtils.isExcludeDatabase(dbName)) {
                    continue;
                }

                try {
                    MongoDatabase database = mongoClient.getDatabase(dbName);

                    // 创建数据库对象
                    SourceAccessDatabase sourceAccessDatabase = new SourceAccessDatabase();
                    sourceAccessDatabase.setDatabaseName(dbName);
                    sourceAccessDatabase.setTables(new HashSet<>());

                    // 获取所有集合（相当于关系数据库中的表）
                    MongoIterable<String> collections = database.listCollectionNames();
                    for (String collectionName : collections) {
                        SourceAccessTable table = new SourceAccessTable();
                        table.setTableName(collectionName);
                        // 注意：MongoDB是文档数据库，没有固定的列结构
                        // 这里暂时设置为空集合，实际使用时可能需要动态分析文档结构
                        table.setColumns(new HashSet<>());
                        sourceAccessDatabase.getTables().add(table);
                    }

                    databases.add(sourceAccessDatabase);
                } catch (Exception e) {
                    // 如果没有权限访问某个数据库，记录日志并跳过该数据库
                    log.warn("无法访问MongoDB数据库 '{}': {}", dbName, e.getMessage());
                    // 可以选择继续处理其他数据库，而不是完全失败
                    continue;
                }
            }
        } catch (Exception e) {
            throw new Exception("获取MongoDB数据库信息失败: " + e.getMessage(), e);
        }

        return databases;
    }
    /**
     * 获取PostgreSQL数据库详细信息，包括表和列
     *
     * @param connection 数据库连接
     * @param sourceAccess 源访问信息
     * @return Set<SourceAccessDatabase> {@link SourceAccessDatabase}
     * @throws SQLException SQL异常
     */
    private Set<SourceAccessDatabase> getPostgreSQLDatabaseDetails(Connection connection, SourceAccess sourceAccess)
            throws SQLException {
        Set<SourceAccessDatabase> databases = new HashSet<>();
        DatabaseMetaData metaData = connection.getMetaData();

        // 首先通过查询pg_database获取真实的数据库列表
        try (Statement stmt = connection.createStatement()) {
            // 查询所有非模板数据库
            ResultSet dbResult = stmt.executeQuery("SELECT datname FROM pg_database WHERE NOT datistemplate");
            while (dbResult.next()) {
                String databaseName = dbResult.getString("datname");
                if (databaseName != null && !ExcludeDatabaseUtils.isExcludeDatabase(databaseName)) {
                    SourceAccessDatabase database = new SourceAccessDatabase();
                    database.setDatabaseName(databaseName);
                    database.setTables(new HashSet<>());
                    databases.add(database);
                }
            }
        } catch (SQLException e) {
            log.warn("无法通过PostgreSQL系统表获取数据库列表，使用默认方法: " + e.getMessage());
            // 如果查询失败，回退到默认方法
            return getAllDatabaseDetails(connection, null);
        }

        // 获取每个数据库中的表和列信息
        for (SourceAccessDatabase database : databases) {
            try {
                // 获取该数据库中的所有表
                ResultSet tables = metaData.getTables(database.getDatabaseName(), "public", "%", new String[]{"TABLE"});
                while (tables.next()) {
                    String tableName = tables.getString("TABLE_NAME");
                    if (tableName != null) {
                        SourceAccessTable table = new SourceAccessTable();
                        table.setTableName(tableName);
                        table.setColumns(new HashSet<>());

                        // 获取表中的所有列
                        ResultSet columns = metaData.getColumns(database.getDatabaseName(), "public", tableName, "%");
                        while (columns.next()) {
                            String columnName = columns.getString("COLUMN_NAME");
                            String columnType = columns.getString("TYPE_NAME");

                            if (columnName != null && columnType != null) {
                                IColumn column = new IColumn();
                                column.setColumnName(columnName);
                                column.setColumnType(columnType);
                                table.getColumns().add(column);
                            }
                        }

                        database.getTables().add(table);
                    }
                }
            } catch (SQLException e) {
                log.warn("无法获取数据库 " + database.getDatabaseName() + " 的表信息: " + e.getMessage());
                // 继续处理其他数据库
            }
        }

        log.info("PostgreSQL数据库信息查询结束");
        return databases;
    }

}
